package com.atguigu.sparkcore.rdd.tranformdoublevalue

import com.atguigu.sparkcore.util.MySparkContextUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * rdd转换-Subtract
  * rdd2.subtract(rdd1)
  * rdd2减去rdd1的值,原来的分区数不会减少
  * 相减完之后会重新分区  subtract(other, partitioner.getOrElse(new HashPartitioner(partitions.length)))，默认情况下使用hashPartition，对分区数进行分区操作
  */
object Subtract {

  def main(args: Array[String]): Unit = {

    // 获取sparkContext
    val sc = MySparkContextUtil.get(args)

    // 创建rdd
    val rdd: RDD[Int] = sc.makeRDD((1 to 10), 2)
    val rdd2: RDD[Int] = sc.makeRDD(Array(1, 3, 5, 7, 9), 2)

    // 输出rdd相关分区信息
    //    rdd.mapPartitionsWithIndex((index, item)=>item.map((index,_))).foreach(println)


    // 转换rdd
    println(rdd.partitioner)
    val result = rdd.subtract(rdd2)

    // 输出
    //    println(result.getNumPartitions)
    result.collect().foreach(println)
    //    result.mapPartitionsWithIndex((index, item)=>item.map((index,_))).foreach(println)

    // 关闭资源
    MySparkContextUtil.close(sc)
  }

}
